package cn.wolfcode.shop.seckillserver.mq;

import cn.wolfcode.shop.common.exception.BusinessException;
import cn.wolfcode.shop.seckillapi.domain.OrderInfo;
import cn.wolfcode.shop.seckillapi.service.ISeckillOrderService;
import cn.wolfcode.shop.seckillapi.vo.SeckillOrderVO;
import cn.wolfcode.shop.seckillserver.util.MQConstants;
import cn.wolfcode.shop.seckillserver.util.OrderMessage;
import cn.wolfcode.shop.seckillserver.web.controller.frontend.SeckillOrderController;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

@Component
public class OrderMQListener {


    @Autowired
    private ISeckillOrderService seckillOrderService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 接受 绑定 ORDER_PEDDING_QUEUE 队列的消息
     * 进行秒杀订单下单的业务操作
     * <p>
     * 成功: 发布 绑定 ORDER_RESULT_EXCHANGE 交换机 携带KEY ORDER.SUCCESS 的消息
     * 失败: 发布 绑定 ORDER_RESULT_EXCHANGE 交换机 携带KEY ORDER.FAIL 的消息
     *
     * @param orderMessage
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queuesToDeclare = {
            @Queue(MQConstants.ORDER_PEDDING_QUEUE)
    })
    public void handlerOrderPeddingQueue(@Payload OrderMessage orderMessage, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        Map<String, Object> map = new HashMap<>();
        map.put("uuid", orderMessage.getUuid());
        map.put("goodId", orderMessage.getGoodId());
        try {

            String orderNo = seckillOrderService.doSeckill((SeckillOrderVO) orderMessage.getData(), Long.parseLong(orderMessage.getUserId()));
            map.put("success", true);
            map.put("orderNo", orderNo);
            rabbitTemplate.convertAndSend(MQConstants.ORDER_RESULT_EXCHANGE, MQConstants.ORDER_RESULT_SUCCESS_KEY, map);

        } catch (BusinessException e) {
            System.out.println("6666666666666");
            e.printStackTrace();
            map.put("success", false);
            map.put("msg", e.getCodeMsg().getMsg());
            rabbitTemplate.convertAndSend(MQConstants.ORDER_RESULT_EXCHANGE, MQConstants.ORDER_RESULT_FAIL_KEY, map);

        } catch (Exception ex) {
            System.out.println("77777777777777");
            ex.printStackTrace();
            map.put("success", false);
            map.put("msg", "抢购失败");
            rabbitTemplate.convertAndSend(MQConstants.ORDER_RESULT_EXCHANGE, MQConstants.ORDER_RESULT_FAIL_KEY, map);

        } finally {

            //手动签收消息
            channel.basicAck(deliveryTag, false);
        }
    }


    /**
     * 接受 绑定 ORDER_RESULT_EXCHANGE 交换机 携带KEY ORDER.FAIL 的消息
     * 对失败订单的数据回滚的业务操作
     * 并发送 绑定 SECKILL_OVER_SIGN_PUBSUB_EX 取消缓存的标记
     *
     * @param map
     * @param deliveryTag
     * @param channel
     * @throws IOException
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(MQConstants.ORDER_RESULT_FAIL_QUEUE),
            exchange = @Exchange(name = MQConstants.ORDER_RESULT_EXCHANGE, type = "topic"),
            key = MQConstants.ORDER_RESULT_FAIL_KEY))
    public void handlerOrderFailMsg(@Payload Map<String, Object> map, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {

        Long goodId = Long.parseLong((String) map.get("goodId"));

        //数据回滚
        seckillOrderService.syncStock(goodId);

        //取消内存的标记
        rabbitTemplate.convertAndSend(MQConstants.SECKILL_OVER_SIGN_PUBSUB_EX, "", goodId);

        //手动签收消息
        channel.basicAck(deliveryTag, false);

    }


    /**
     * 创建延迟消息队列 Bean
     *
     * @return
     */
    @Bean
    public org.springframework.amqp.core.Queue orderDelayQueue() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", MQConstants.DELAY_EXCHANGE);
        arguments.put("x-dead-letter-routing-key", MQConstants.ORDER_DELAY_KEY);
        arguments.put("x-message-ttl", 1000 * 60 * 15);
        org.springframework.amqp.core.Queue queue = new org.springframework.amqp.core.Queue(MQConstants.ORDER_RESULT_SUCCESS_DELAY_QUEUE, true, false, false, arguments);
        return queue;
    }

    @Bean
    public Binding binding1a(org.springframework.amqp.core.Queue orderDelayQueue) {
        return BindingBuilder.bind(orderDelayQueue)
                .to(new TopicExchange(MQConstants.ORDER_RESULT_EXCHANGE))
                .with(MQConstants.ORDER_RESULT_SUCCESS_KEY);
    }


    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(MQConstants.ORDER_TIMEOUT_QUEUE),
            exchange = @Exchange(name = MQConstants.DELAY_EXCHANGE, type = "topic"),
            key = MQConstants.ORDER_DELAY_KEY))
    public void handlerTimeOutOrder(@Payload Map<String, Object> map, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {

        try {

            String orderNo = (String) map.get("orderNo");
            Long goodId = Long.parseLong((String) map.get("goodId"));

            seckillOrderService.timeOutCancelOrder(orderNo, goodId);

        } catch (Exception e) {

            e.printStackTrace();
        }

        channel.basicAck(deliveryTag, false);
    }


    @RabbitListener(bindings = {@QueueBinding(
            value = @Queue,
            exchange = @Exchange(value = MQConstants.SECKILL_OVER_SIGN_PUBSUB_EX, type = "fanout"))})
    public void handlerCancelSeckillOverSign(@Payload Long goodId, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
        SeckillOrderController.overStock.put(goodId, false);
        channel.basicAck(deliveryTag, false);
    }
}